package com.zhangpan.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.zhangpan.realtime.common.bean.TableProcessDim;

import com.zhangpan.realtime.common.constant.Constant;
import com.zhangpan.realtime.common.util.HBaseUtil;
import com.zhangpan.realtime.common.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.io.IOException;

/**
 * @author zhangpan
 */
@Slf4j
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {


    private Connection conn;

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConn(conn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> t, Context ctx) throws Exception {

        log.info("HbaseSink进来了");
        TableProcessDim tp = t.f1;
        JSONObject data = t.f0;
        log.info("当前数据:{}",data);
        String opType = data.getString("op_type");
        if ("delete".equals(opType)) {
            // 删除纬度信息
            log.info("删除纬度");
            delDim(t);
        } else {
            log.info("添加纬度");
            putDim(t);
        }

        // 如果纬度是update 或delete 则删除缓存中的纬度数据
        if("delete".equals(opType) || "update".equals(opType)) {
            String key = RedisUtil.getKey(tp.getSinkTable(),data.getString("id"));
            jedis.del(key);
        }
    }

    private void putDim(Tuple2<JSONObject,TableProcessDim> t) throws IOException {

        System.out.println("写入到hbase中。。。。");
        JSONObject data = t.f0;
        TableProcessDim tp = t.f1;
        String rowKey = data.getString(tp.getSinkRowKey());
        JSONObject dataCopy = new JSONObject();
        dataCopy.putAll(data);
        dataCopy.remove("op_type");
        HBaseUtil.putRow(conn, Constant.HBASE_NAMESPACE,tp.getSinkTable(),rowKey,tp.getSinkFamily(),dataCopy);

    }

    private void delDim(Tuple2<JSONObject,TableProcessDim> t) throws IOException {
        JSONObject data = t.f0;
        TableProcessDim tp = t.f1;
        String rowKey = data.getString(tp.getSinkRowKey());
        HBaseUtil.delRow(conn,Constant.HBASE_NAMESPACE,tp.getSinkTable(),rowKey);

    }
}
